import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.Code;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.WatchedEvent;

import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;

/**
 * The JobTrackerNewJobThread class
 *
 * @author  Joshua Kwan
 */

public class JobTrackerRecoveryThread extends JobTrackerNewJobThread {
		
	public JobTrackerRecoveryThread (ZkConnector zkc, String userID, String hash){
		super (zkc, userID, hash);
		System.out.println("JobTrackerRecoveryThread created for " + userID + " to get password for " + hash);
		this.userID = userID;
		this.hash = hash;
		jobPath = JOBPOOL_PATH + "/" + userID + "/" + hash;
		this.zkc = zkc;
        zk = zkc.getZooKeeper();
	}
	
	public void run() {
		
		try {
			//recover data about incomplete tasks and workers
			recoverAssignments();
			
			//create a watcher for the job
			jobWatcher = new Watcher(){
				// @Override
	        	public void process(WatchedEvent event) {
	        		handleJobEvent(event);
	        	}
			};
			checkPath(jobPath, jobWatcher);
	        
			//create the watcher for the worker pool
			wpWatcher = new Watcher(){
	        	//@Override
	        	public void process(WatchedEvent event) {
	        		handleWorkerPoolEvent(event);
	        	}
	        };
			//get the children of the workerpool
	        List<String> wpList = zk.getChildren(WORKERPOOL_PATH, wpWatcher);
	        updateWorkerMap(wpList);
	        
	        //try to assign all unassigned tasks to free workers
	        initialAssignment();
	        
	        while (!jobFinished) {
	        	Thread.sleep(SLEEPTIME);
	        	for(String wp: workerMap.keySet()){
	        		checkWorker(wp);
	        	}
	        }
			
		} catch (KeeperException e) {
			System.out.println(e);
			if (e.code() == KeeperException.Code.NODEEXISTS){
				System.out.println("A job for " + hash + " has already been requested.");
				//exit?
			}
		} catch (InterruptedException e) {
			System.out.println(e);
			//try again?...
		} catch (IllegalArgumentException e) {
			System.out.println(e);
			//maybe exit??
		}

	}
    
	private void recoverAssignments() throws KeeperException, InterruptedException {
		//get list of tasks - children of the job
		List<String> incompleteTasks = zk.getChildren(jobPath, false);
		//Traverse list of tasks to see if there are workers working on it
		ListIterator i = incompleteTasks.listIterator();
		while (i.hasNext()) {
			String taskPath = jobPath + "/" + i.next().toString();
			Stat stat = new Stat();
			byte[] taskData = zk.getData(taskPath, false, stat);
			if (taskData != null) {
				String[] strTaskData = new String(taskData).split(ZkConnector.DATADELIMITER);
				if (strTaskData.length == 3) {
					if (zkc.exists(strTaskData[2], false) != null) {
						workerTask.put(strTaskData[2], taskPath);
					} else {
						//worker also failed, remove the taskPath and add the task to the unassignedTasks
						atomicRemoveTaskPath(strTaskData[2], taskPath);
						unassignedTasks.add(taskPath);
					}
				} else {
					//add the task to unassignedTasks
					unassignedTasks.add(taskPath);
				}
			}
		}
		
	}

}